package cluster.front

import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger

import akka.actor.{Actor, ActorRef, ActorSystem, Props, Terminated}
import akka.pattern.Patterns
import akka.util.Timeout
import cluster.{BackendRegistration, JobFailed, TransformationJob}
import com.typesafe.config.ConfigFactory

import scala.concurrent.Await
import scala.concurrent.duration.FiniteDuration

/**
  * Created by shouf on 2016/5/18.
  */
class TransformationFrontend extends Actor {
  var backEnds = IndexedSeq.empty[ActorRef]
  var jobCounter = 0
  def receive={
    case job: TransformationJob if backEnds.isEmpty =>
      sender() ! JobFailed("Service unavailable, try again later", job)

    case job: TransformationJob =>
      jobCounter += 1
      backEnds(jobCounter % backEnds.size) forward job // forward  只是向前传递消息，但是消息的发送者还是最原始的sender，即发给这条信息的sender  此处的意思是front只是作为一个传递数据的工具actor

    case BackendRegistration if !(backEnds.contains(sender())) =>
      context watch sender()
      backEnds = backEnds :+ sender()

    case Terminated(a) =>
      println(s"${a.path} is Terminated")
      backEnds = backEnds.filterNot(_ == a)
  }
}

object TransformationFrontend extends App {
  val port = if (args.isEmpty) 0 else args(0).toInt
  val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port").
    withFallback(ConfigFactory.parseString("akka.cluster.roles = [frontend]")).
    withFallback(ConfigFactory.load.getConfig("cluster"))

  val system = ActorSystem("clusterSys", config)
  val frontend = system.actorOf(Props[TransformationFrontend], "frontend")

  val counter = new AtomicInteger()
  import scala.concurrent.ExecutionContext.Implicits.global
  system.scheduler.schedule(FiniteDuration(2, TimeUnit.SECONDS), FiniteDuration(2, TimeUnit.SECONDS)) {
    implicit val timeout = Timeout(5, TimeUnit.SECONDS)
    val pattern = Patterns.ask(frontend, TransformationJob("hello-" + counter.incrementAndGet()), timeout) //使用等待的方式发送消息并得到结果
    println(Await.result(pattern, timeout.duration))
  }
}
